package com.gjy.kafka.resolve.producer;

import com.gjy.kafka.resolve.config.KafkaProducerConfig;
import lombok.Getter;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 15:19:41
 */
@Getter
public class TransactionalProducer implements AutoCloseable {

    private static final Logger logger = LoggerFactory.getLogger(TransactionalProducer.class);
    private final Producer<String, String> producer;

    public TransactionalProducer(String transactionalId) {
        Properties props = KafkaProducerConfig.buildConfig(transactionalId);
        this.producer = new KafkaProducer<>(props);
        this.producer.initTransactions(); // 必须先 init
    }

    /**
     * 在一次事务内发送多条记录
     */
    public void sendInTransaction(String topic, Iterable<ProducerRecord<String, String>> records) {
        try {
            producer.beginTransaction();
            for (ProducerRecord<String, String> rec : records) {
                producer.send(rec);
            }
            producer.commitTransaction();
            logger.info("事务提交成功，topic={}", topic);
        } catch (Exception e) {
            logger.error("事务提交失败，执行回滚", e);
            producer.abortTransaction();
        }
    }

    @Override
    public void close() {
        producer.flush();
        producer.close();
    }

}
